package com.powerwyx.shortlink.project.mq.listener;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.powerwyx.shortlink.project.common.convention.exception.ServiceException;
import com.powerwyx.shortlink.project.mapper.*;
import com.powerwyx.shortlink.project.mq.idempotent.MessageQueueIdempotent;
import com.powerwyx.shortlink.project.pojo.dto.ShortLinkStats;
import com.powerwyx.shortlink.project.pojo.entity.*;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.powerwyx.shortlink.project.common.constant.MQConstant.SHORT_LINK_STATS_EXCHANGE;
import static com.powerwyx.shortlink.project.common.constant.MQConstant.SHORT_LINK_STATS_KEY;
import static com.powerwyx.shortlink.project.common.constant.RedisConstant.LOCK_GID_UPDATE_KEY;
import static com.powerwyx.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

/**
 * @author 王艺锡
 * @version 1.0
 */
@Component
@Slf4j
public class ShortLinkStatsListener {
    @Autowired
    private LinkMapper linkMapper;

    @Autowired
    private LinkGotoMapper linkGotoMapper;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private LinkAccessStatsMapper linkAccessStatsMapper;

    @Autowired
    private LinkLocalStatsMapper linkLocalStatsMapper;

    @Autowired
    private LinkBrowserStatsMapper linkBrowserStatsMapper;

    @Autowired
    private LinkAccessLogsMapper linkAccessLogsMapper;

    @Autowired
    private LinkOsStatsMapper linkOsStatsMapper;

    @Autowired
    private LinkDeviceStatsMapper linkDeviceStatsMapper;

    @Autowired
    private LinkNetworkStatsMapper linkNetworkStatsMapper;

    @Autowired
    private LinkStatsTodayMapper linkStatsTodayMapper;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private MessageQueueIdempotent messageQueueIdempotent;

    @Value("${short-link.stats.local.amap-key}")
    private String statsLocalAmapKey;

    @Transactional(rollbackFor = Exception.class)
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "short_link_stats_queue", durable = "true"),
            exchange = @Exchange(name = SHORT_LINK_STATS_EXCHANGE, durable = "true", type = "direct"),
            key = {SHORT_LINK_STATS_KEY}
    ))
    public void shortLinkStatistics(@Payload ShortLinkStats shortLinkStats, @Header("amqp_messageId") String messageId) {
        //查询缓存，判断当前消息是否已经消费过
        //如果已经消费过
        if(!messageQueueIdempotent.isConsumed(messageId)){
            //判断业务流程是否已经完成
            if(messageQueueIdempotent.isAccomplish(messageId)){
                //如果业务流程已经完成，代表当前消息是一个重复消息，直接返回即可
                return;
            }
            //如果业务流程还未完成，抛异常等待mq重新投递消息，等key过期后即可正常消费消息
            throw new ServiceException("消息消费中 但是未完成!");
        }

        try {
            String fullShortUrl = shortLinkStats.getFullShortUrl();
            //添加读锁，防止添加监控数据时分组被修改导致数据不一致的问题
            RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
            RLock rLock = readWriteLock.readLock();
            rLock.lock();

            try {
                AtomicBoolean uvFlag = shortLinkStats.getUvFlag();
                Boolean uipFlag = shortLinkStats.getUipFlag();
                String remoteAddr = shortLinkStats.getRemoteAddr();
                Date date = shortLinkStats.getCurrentDate();

                LinkGoto linkGoto = linkGotoMapper.selectOne(new QueryWrapper<LinkGoto>()
                        .eq("full_short_url", fullShortUrl));

                String gid = linkGoto.getGid();

                //获取当前的小时数
                int hour = DateUtil.hour(date, true);
                //获取当前是星期几
                Week week = DateUtil.dayOfWeekEnum(date);
                int weekValue = week.getIso8601Value();

                //构建对象,添加基础监控数据
                LinkAccessStats linkAccessStats = LinkAccessStats.builder()
                        .pv(1)
                        .uv(uvFlag.get() ? 1 : 0)
                        .uip(uipFlag ? 1 : 0)
                        .hour(hour)
                        .weekday(weekValue)
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .build();

                linkAccessStatsMapper.shortLinkStatistics(linkAccessStats);

                //构建对象,添加地区监控数据
                Map<String, Object> localParamMap = new HashMap<>();
                localParamMap.put("key", statsLocalAmapKey);
                localParamMap.put("ip", remoteAddr);
                String localResultStr = HttpUtil.get(AMAP_REMOTE_URL, localParamMap);
                JSONObject localResultObj = JSONUtil.parseObj(localResultStr);
                String infoCode = localResultObj.getStr("infocode");
                String actualProvince = "未知";
                String actualCity = "未知";
                if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                    String province = localResultObj.getStr("province");
                    boolean unKnowFlag = StrUtil.equals(province, "[]");
                    LinkLocalStats linkLocalStats = LinkLocalStats.builder()
                            .fullShortUrl(fullShortUrl)
                            .gid(gid)
                            .date(date)
                            .cnt(1)
                            .province(actualProvince = unKnowFlag ? "未知" : province)
                            .city(actualCity = unKnowFlag ? "未知" : localResultObj.getStr("city"))
                            .adcode(unKnowFlag ? "未知" : localResultObj.getStr("adcode"))
                            .country("中国")
                            .build();
                    linkLocalStatsMapper.shortLinkLocalStatistics(linkLocalStats);
                }

                //构建对象,添加操作系统访问量监控
                LinkOsStats linkOsStats = LinkOsStats.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .os(shortLinkStats.getOs())
                        .build();
                linkOsStatsMapper.shortLinkOsStatistics(linkOsStats);

                //构建对象,添加浏览器访问量监控
                LinkBrowserStats linkBrowserStats = LinkBrowserStats.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .cnt(1)
                        .browser(shortLinkStats.getBrowser())
                        .build();
                linkBrowserStatsMapper.shortLinkBrowserStatistics(linkBrowserStats);

                //构建对象,添加设备访问量监控
                LinkDeviceStats linkDeviceStats = LinkDeviceStats.builder()
                        .device(shortLinkStats.getDevice())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .build();
                linkDeviceStatsMapper.shortLinkDeviceStatistics(linkDeviceStats);

                //构建对象,添加网络访问量监控
                LinkNetworkStats linkNetworkStats = LinkNetworkStats.builder()
                        .network(shortLinkStats.getNetwork())
                        .cnt(1)
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .build();
                linkNetworkStatsMapper.shortLinkNetworkStatistics(linkNetworkStats);

                //构建对象,统计高频IP访问
                LinkAccessLogs linkAccessLogs = LinkAccessLogs.builder()
                        .user(shortLinkStats.getUserFlag())
                        .ip(remoteAddr)
                        .browser(shortLinkStats.getBrowser())
                        .os(shortLinkStats.getOs())
                        .network(shortLinkStats.getNetwork())
                        .device(shortLinkStats.getDevice())
                        .local(StrUtil.join("-", "中国", actualProvince, actualCity))
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .build();
                linkAccessLogsMapper.insert(linkAccessLogs);

                linkMapper.incrementStats(gid, fullShortUrl, 1, uvFlag.get() ? 1 : 0, uipFlag ? 1 : 0);

                LinkStatsToday linkStatsToday = LinkStatsToday.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(date)
                        .todayPv(1)
                        .todayUv(uvFlag.get() ? 1 : 0)
                        .todayUip(uipFlag ? 1 : 0)
                        .build();
                linkStatsTodayMapper.shortLinkTodayStatistics(linkStatsToday);
            } catch (Throwable ex) {
                throw ex;
            } finally {
                rLock.unlock();
            }
        } catch (Exception ex) {
            //如果业务出现错误，删除标识
            messageQueueIdempotent.delIdempotent(messageId);
            log.error("短链接访问量统计异常", ex);
            throw ex;
        }
        //消息消费成功，设置幂等标识为已完成
        messageQueueIdempotent.setAccomplish(messageId);

    }
}
